package com.shu;

import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.StormSubmitter;
import org.apache.storm.spout.Scheme;
import org.apache.storm.topology.TopologyBuilder;
import org.apache.storm.tuple.Fields;

import com.rabbitmq.client.ConnectionFactory;
import com.shu.bolt.PrintOutput;
import com.shu.bolt.SplitWordBolt;
import com.shu.bolt.WordCountBolt;
import com.shu.scheme.JobCustomMessageScheme;

import io.latent.storm.rabbitmq.RabbitMQSpout;
import io.latent.storm.rabbitmq.config.ConnectionConfig;
import io.latent.storm.rabbitmq.config.ConsumerConfig;
import io.latent.storm.rabbitmq.config.ConsumerConfigBuilder;

public class WordCountCountTopology {

	public static void main(String[] args) throws Exception {

		Scheme scheme = new JobCustomMessageScheme();
		RabbitMQSpout spout = new RabbitMQSpout(scheme);

		ConnectionConfig connectionConfig = new ConnectionConfig("192.168.1.15", 5672, "eagle", "eagle",
				ConnectionFactory.DEFAULT_VHOST, 10); // host, port, username, password, virtualHost, heartBeat
		ConsumerConfig spoutConfig = new ConsumerConfigBuilder().connection(connectionConfig)
				.queue("com.mzj.storm.test").prefetch(200).requeueOnFail().build();

		TopologyBuilder builder = new TopologyBuilder();

		builder.setSpout("storm-spout", spout,2).setNumTasks(5).addConfigurations(spoutConfig.asMap()).setMaxSpoutPending(200);
		// 读取spout里的数据，进行split处理
		builder.setBolt("splitBolt", new SplitWordBolt(), 3).shuffleGrouping("storm-spout");
		// 读取split后的数据，进行count处理
		builder.setBolt("countBolt", new WordCountBolt(), 15).setNumTasks(5).fieldsGrouping("splitBolt", new Fields("word"));
		// 保存计算结果
		builder.setBolt("saveBolt", new PrintOutput(), 3).fieldsGrouping("countBolt", new Fields("word"));

		//远程提交
		Config conf = new Config();
		conf.setDebug(true);
		// 有参数则到集群跑，没有则在本地跑
		if (args != null && args.length > 0) {
			conf.setNumWorkers(2);
			StormSubmitter.submitTopology(args[0], conf, builder.createTopology());
		} else {
			LocalCluster cluster = new LocalCluster();
			cluster.submitTopology("word-split", conf, builder.createTopology());
			Thread.sleep(300000); // 5分钟后自动结束
			cluster.shutdown();
		}

	}

}
